﻿
using System;
using System.Text;
using System.Collections.Generic;
using System.Net;
using System.Net.Sockets;
using System.Linq;

using LExperiment.NetworkLib;

namespace LExperiment.Master
{
    sealed class LMaster : IDisposable
    {
        // ========================================================== //
        #region [ member variables ]

        private object m_lock = new object();

        /// <summary>
        /// tasks dispatched to the same computers share the same channel
        /// no need to build the channel again
        /// Key = IP address and port of the computer where the subscriber daemon is running
        /// Value = TCP channel connected with the subscriber daemon
        /// </summary>
        private IDictionary<IPEndPoint, LTcpChannel> m_cacheChannels;

        /// <summary>
        /// Key = subscriber id
        /// Value = TCP channel connected with the subscriber daemon which launch the subscriber
        /// </summary>
        private IDictionary<int, LTcpChannel> m_subscribers;

        private bool m_disposed;

        private string m_pubAddress;
        private int m_pubPort;

        #endregion

        // ========================================================== //
        #region [ constructor and destructor ]

        public LMaster(string pubAddress, int port)
        {
            m_cacheChannels = new Dictionary<IPEndPoint, LTcpChannel>();
            m_subscribers = new Dictionary<int, LTcpChannel>();

            m_pubAddress = pubAddress;
            m_pubPort = port;
        }

        ~LMaster() { Dispose(false); }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        /// <summary>
        /// no synchronization, since after disposed, the object should not be used again
        /// </summary>
        private void Dispose(bool isDisposing)
        {
            if (m_disposed)
                return;

            try
            {
                if (isDisposing)
                {
                    // though LTcpChannels are stored in two containers
                    // dispose them in one container is enough
                    foreach (KeyValuePair<IPEndPoint, LTcpChannel> kv in m_cacheChannels)
                    {
                        kv.Value.Dispose();
                    }
                    m_cacheChannels.Clear();
                    m_subscribers.Clear();
                }
            }
            finally
            {
                m_disposed = true;
            }
        }

        #endregion

        // ========================================================== //
        private LTcpChannel GetChannel(string address, int port)
        {
            IPEndPoint remoteEndpoint = NetworkUtility.CreateEndPoint(address, port);
            if (m_cacheChannels.ContainsKey(remoteEndpoint))
                return m_cacheChannels[remoteEndpoint];
            else
            {
                Socket socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
                socket.Connect(remoteEndpoint);
                Console.WriteLine("[+] build connection with daemon <{0}>", remoteEndpoint.ToString());

                LTcpChannel channel = new LTcpChannel(socket);
                m_cacheChannels.Add(remoteEndpoint, channel);
                System.Diagnostics.Debug.Assert(remoteEndpoint.Equals(channel.RemoteEndpoint));

                channel.BeginReceive();// must be invoked to be notified when passively closed by remote peer
                channel.PassiveSockClosedEvtHandler += this.OnChannelClosed;
                return channel;
            }
        }

        public void Start(string address, int port, IEnumerable<int> subscriberIds)
        {
            lock (m_lock)
            {
                LTcpChannel channel = GetChannel(address, port);

                // ---------------------------------- prepare command
                StartCommand startcmd = new StartCommand();
                startcmd.PubAddress = m_pubAddress;
                startcmd.PubPort = m_pubPort;

                foreach (int id in subscriberIds)
                {
                    startcmd.AddSubscriberId(id);
                    m_subscribers.Add(id, channel);
                    Console.WriteLine("[+] subscriber<{0}> added", id.ToString());
                }

                // ---------------------------------- send command out
                channel.BeginSend(new PacketHeader { Cmd = CmdConstant.Start }, startcmd.BinaryImage);
            }
        }

        public void Stop(IEnumerable<int> subscriberIds)
        {
            lock (m_lock)
            {
                IDictionary<IPEndPoint, StopCommand> stopCmds = new Dictionary<IPEndPoint, StopCommand>();
                foreach (int id in subscriberIds)
                {
                    if (m_subscribers.ContainsKey(id))
                    {
                        StopCommand cmd = GetStopCommand(stopCmds, id);
                        cmd.AddSubscriberId(id);

                        m_subscribers.Remove(id);
                        Console.WriteLine("[-] subscriber<{0}> removed", id.ToString());
                    }
                }

                PacketHeader header = new PacketHeader { Cmd = CmdConstant.Stop };
                foreach (KeyValuePair<IPEndPoint, StopCommand> kv in stopCmds)
                {
                    LTcpChannel channel = m_cacheChannels[kv.Key];
                    channel.BeginSend(header, kv.Value.BinaryImage);
                    Console.WriteLine("[-] stop subscribers on daemon<{0}>", kv.Key.ToString());
                }
            }
        }

        private StopCommand GetStopCommand(IDictionary<IPEndPoint, StopCommand> stopCmds, int subscriberId)
        {
            LTcpChannel channel = m_subscribers[subscriberId];

            if (stopCmds.ContainsKey(channel.RemoteEndpoint))
                return stopCmds[channel.RemoteEndpoint];
            else
            {
                StopCommand cmd = new StopCommand();
                stopCmds.Add(channel.RemoteEndpoint, cmd);
                return cmd;
            }
        }

        private void OnChannelClosed(object source, EventArgs evtargs)
        {
            LTcpChannel channel = (LTcpChannel)source;

            IList<int> closedSubIds = new List<int>();
            lock (m_lock)
            {
                m_cacheChannels.Remove(channel.RemoteEndpoint);

                IDictionary<int, LTcpChannel> oriSubscribers = m_subscribers;
                // create a new container, and re-insert, to avoid "remove while iteration"
                m_subscribers = new Dictionary<int, LTcpChannel>();

                foreach (KeyValuePair<int, LTcpChannel> kv in oriSubscribers)
                {
                    if (!object.ReferenceEquals(kv.Value, channel))
                    {
                        m_subscribers.Add(kv.Key, kv.Value);
                    }
                    else
                    {
                        closedSubIds.Add(kv.Key);
                    }
                }
            }// lock

            Console.WriteLine("[!] connection with <{0}> is closed", channel.RemoteEndpoint.ToString());

            StringBuilder strBuilder = new StringBuilder();
            closedSubIds.Aggregate(strBuilder, (sb, number) => sb.Append(string.Format("{0} ", number.ToString())));
            Console.WriteLine("[!] subscribers [ {0}] are killed", strBuilder.ToString());

            channel.Dispose();
        }
    }
}